package com.chukun.kafa.action.producer;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.errors.AuthorizationException;
import org.apache.kafka.common.errors.OutOfOrderSequenceException;
import org.apache.kafka.common.errors.ProducerFencedException;

import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;

/**
 * kafka事务生产者
 * @author 初坤
 *
 */
public class KafkaTransactionalProducer {

	
	public static void main(String[] args) throws InterruptedException, ExecutionException {
		
		Properties props = KafkaPropertiesUtils.transactionProps("test_transactional_producer");
		KafkaProducer<String, String> producer = new KafkaProducer<>(props);
		producer.initTransactions();
		try {
            producer.beginTransaction();
			for (int i = 0; i < 5; i++) {
				Future<RecordMetadata> send = producer.send(new ProducerRecord<>("test_transaction_producer", Integer.toString(i), "test_transaction_producer_message_" + i));
				System.out.println("kafka offset ----> " + send.get().offset());
				TimeUnit.MILLISECONDS.sleep(1000L);
			}
			producer.commitTransaction();
		} catch (ProducerFencedException | OutOfOrderSequenceException | AuthorizationException e) {
			producer.close();
		} catch (KafkaException e) {
			producer.abortTransaction();
		}
		producer.close();
	}
	
}
